package org.cainiao.basic.reactive.create;

import org.cainiao.basic.reactive.source.*;
import reactor.core.publisher.Flux;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class CreateTest {

    private static MyEventProcessor<String> myEventProcessor;
    private static MyMessageProcessor<String> myMessageProcessor;
    private static Channel channel;

    public static void main(String[] args) {
        test();
    }

    public static void test1() {
        Flux<String> flux = Flux.generate(
            () -> 0,
            (state, sink) -> {
                sink.next("3 x " + state + " = " + 3 * state);
                if (state == 10) {
                    sink.complete();
                }
                return state + 1;
            });
    }

    public static void test2() {
        Flux<String> flux = Flux.generate(
            AtomicLong::new,
            (state, sink) -> {
                long i = state.getAndIncrement();
                sink.next("3 x " + i + " = " + 3 * i);
                if (i == 10) {
                    sink.complete();
                }
                return state;
            });
    }

    public static void test3() {
        Flux<String> flux = Flux.generate(
            AtomicLong::new,
            (state, sink) -> {
                long i = state.getAndIncrement();
                sink.next("3 x " + i + " = " + 3 * i);
                if (i == 10) {
                    sink.complete();
                }
                return state;
            },
            (state) -> System.out.println("state: " + state));
    }

    public static void test4() {
        Flux<String> bridge = Flux.create(sink -> myEventProcessor.register(
            new MyEventListener<>() {
                public void onDataChunk(List<String> chunk) {
                    for (String s : chunk) {
                        sink.next(s);
                    }
                }

                public void processComplete() {
                    sink.complete();
                }
            }));
    }

    public static void test5() {
        Flux<String> bridge = Flux.push(sink -> myEventProcessor.register(
            new SingleThreadEventListener<>() {
                public void onDataChunk(List<String> chunk) {
                    for (String s : chunk) {
                        sink.next(s);
                    }
                }

                public void processComplete() {
                    sink.complete();
                }

                public void processError(Throwable e) {
                    sink.error(e);
                }
            }));
    }

    public static void test6() {
        Flux<String> bridge = Flux.create(sink -> {
            myMessageProcessor.register(
                messages -> {
                    for (String s : messages) {
                        sink.next(s);
                    }
                });
            sink.onRequest(n -> {
                List<String> messages = myMessageProcessor.getHistory(n);
                for (String s : messages) {
                    sink.next(s);
                }
            });
        });
    }

    public static void test() {
        Flux<String> bridge = Flux.create(sink -> sink
            .onRequest(n -> channel.poll(n))
            .onCancel(() -> channel.cancel())
            .onDispose(() -> channel.close()));
    }

    public static void setMyEventProcessor(
        MyEventProcessor<String> myEventProcessor, MyMessageProcessor<String> myMessageProcessor, Channel channel) {
        CreateTest.myEventProcessor = myEventProcessor;
        CreateTest.myMessageProcessor = myMessageProcessor;
        CreateTest.channel = channel;
    }

}
